/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.cst.lls.util;

import cn.ac.iie.di.cst.lls.common.RuntimeEnv;
import static cn.ac.iie.di.cst.lls.common.RuntimeEnv.changeColumns;
import static cn.ac.iie.di.cst.lls.common.RuntimeEnv.docsReader;
import static cn.ac.iie.di.cst.lls.common.RuntimeEnv.docsSchema;
import static cn.ac.iie.di.cst.lls.common.RuntimeEnv.mqschema;
import static cn.ac.iie.di.cst.lls.common.RuntimeEnv.schemaDatumReader;
import cn.ac.iie.di.cst.lls.se.CstdxStartup;
import cn.ac.iie.util.DataEncryption;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.logging.Level;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.log4j.Logger;

/**
 *
 * @author wanglin
 */
public class DecodeMessage implements Runnable {

    private static org.apache.log4j.Logger logger = Logger.getLogger(DecodeMessage.class.getName());


    @Override
    public void run() {
        while (true) {
            try {
                DataInfo msgData = CstdxStartup.ab.take();
//                logger.info(Thread.currentThread().getId() + " receives message " + msgData.getTopic() + " at " + System.nanoTime()+" ");
//                String mq = msg.getTopic();
//                    String mqs = mq.replace("_mq", "");
                Schema docSchema = mqschema.get(msgData.getTopic().replace("_mq", ""));
                if (docSchema == null) {
                    logger.error("docSchema is null|| mqschemasis " + mqschema.toString() + "  ||docschema is :" + msgData.getTopic());
                    System.exit(0);
                }
                ByteArrayInputStream docsbis = new ByteArrayInputStream(msgData.getBody());
                BinaryDecoder docsbd = new DecoderFactory().binaryDecoder(docsbis, null);
                GenericRecord docsRecord = new GenericData.Record(docsSchema);

                docsReader.read(docsRecord, docsbd);
                GenericArray docsSet = (GenericData.Array<GenericRecord>) docsRecord.get("doc_set");

                if (docsSet.size() < 1) {
                    logger.warn("doc set is empty");
                } else {
                    Iterator itor = docsSet.iterator();
                    GenericDatumReader<GenericRecord> dosReader = (GenericDatumReader<GenericRecord>) schemaDatumReader.get(docSchema);
                    while (itor.hasNext()) {
                        ByteArrayInputStream databis = new ByteArrayInputStream(((ByteBuffer) itor.next()).array());
                        BinaryDecoder dataDecoder = new DecoderFactory().binaryDecoder(databis, null);
                        GenericRecord record = new GenericData.Record(docSchema);
                        dosReader.read(record, dataDecoder);
                        for (int q = 0; q < changeColumns.length; q++) {
                            try {
                                String newc = record.get(RuntimeEnv.changeColumns[q]).toString();
                                String newcolumn = DataEncryption.encrypt(newc);
                                record.put(changeColumns[q], newcolumn);
                            } catch (Exception E) {
                                logger.error(E);
                            }
                        }
                        while (true) {
//                            boolean b=RuntimeEnv.mq2queue.get(msgData.getTopic().replace("_mq", "")).offer(record);
//                            System.out.println(b);
                            if (!RuntimeEnv.mq2queue.get(msgData.getTopic().replace("_mq", "")).offer(record)) {
                                try {
                                    Thread.sleep(400);
//                                    logger.info(" offer data buffer" + msgData.getTopic());
                                } catch (Exception ex) {
                                }
                            } else {
                                break;
                            }
                        }
                    }
                }
            } catch (Exception ex) {
                ex.printStackTrace();
                logger.error(ex);
                logger.error(ex.getMessage(),ex);
            }
        }
    }
}
